package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class IsNewprocessFuncation extends KeyedProcessFunction<String,String,String> {
    //声明变量
    transient ValueState<String> firstDateState=null;

    // 初始化变量
    @Override
    public void open(Configuration parameters) throws Exception {
        firstDateState=getRuntimeContext().getState(new ValueStateDescriptor<String>("first-date",String.class));
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {

        //  获取状态
        String firstDate = firstDateState.value();

        //解析字段
        String isNewValue = JSON.parseObject(value).getJSONObject("common").getString("is_new");
        Long tsValue = JSON.parseObject(value).getLong( "ts");

        if("1".equals(isNewValue)){
            if (null==firstDate){
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            }
            else {
                if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                    JSON.parseObject(value).getJSONObject("common").put("is_new","0");

                }
            }
        }
        else {
            if (null == firstDate){
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue-8640000000L,DateTimeUtil.DATE_FORMAT));
            }
        }
out.collect(value);
    }
}
